#coding:utf8

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, StructField
import pyspark.sql.functions as F

# 电影评分，需求如下：
#     1.查询用户平均分
#     2.查询电影平均分
#     3.查询大于平均分的电影的数量
#     4.查询高分电影中(>3)打分次数最多的用户，并求出此人打的平均分
#     5.查询每个用户的平均分，最低分、最高分
#     6.查询被评分超过100次的电影的平均分、排名TOP100
if __name__ == '__main__':
    # 构建SparkSession执行环境入口对象
    spark = SparkSession.builder.\
        appName("test_parque").\
        master("local[*]").\
        getOrCreate()

    # 定义数据结构
    struct = StructType().add("user_id",StringType()).\
        add("movie_id",StringType()). \
        add("rank", IntegerType()). \
        add("create_time", IntegerType())

    # 读取数据
    df = spark.read.format("csv").\
        option("sep","\t").\
        option("header",False).\
        option("encoding","utf-8").\
        schema(schema=struct).\
        load('../data/input/u.data')

    # 1.查询用户平均分
    df.groupBy('user_id').avg('rank').\
    withColumnRenamed("avg(rank)","avg_rank"). \
    withColumn('avg_rank', F.round('avg_rank',2)).\
    orderBy('avg_rank',ascending=False).\
    show()

    # 2.查询电影平均分
    df.createTempView('movie_table')
    spark.sql("select movie_id,round(avg(rank),2) avg_rank from movie_table group by movie_id order by avg(rank) desc").show()

    # 3.查询大于平均分的电影的数量
    print("大于平均分的电影的数据量：",df.where(df['rank']>df.select(F.avg(df['rank'])).first()['avg(rank)']).count())

    # 4.查询高分电影中( > 3)打分次数最多的用户，并求出此人打的平均分
    # 先找出这个人
    user_id = df.where(df["rank"]>3).\
        groupBy("user_id").\
        count().\
        orderBy("count",ascending=False).\
        first()["user_id"]
    print(user_id)
    # 根据这个人进行筛选后求平均
    df.where(df['user_id']==user_id).\
        select(F.round(F.avg(df['rank']),2)).show()

    # 5.查询每个用户的平均分，最低分、最高分
    # 这里对多个字段统计聚合结果，需要用到agg函数
    df.groupBy('user_id').\
        agg(
        F.round(F.avg(df['rank'])).alias('avg_rank'),
        F.max(df['rank']).alias('max_rank'),
        F.min(df['rank']).alias('min_rank')
    ).show()

    # 6.查询被评分超过100次的电影的平均分、排名TOP100
    df.groupBy('movie_id').\
        agg(
        F.count(df['movie_id']).alias('cnt'),
        F.round(F.avg(df['rank']),2).alias('avg_rank')
    ).where('cnt > 100').\
        orderBy('avg_rank',ascending=False).\
        limit(100).show()